package com.ds.infrastructure.hyperspace.container.endpoint.destination;

import com.ds.infrastructure.hyperspace.common.EndpointNode;
import com.ds.infrastructure.hyperspace.common.TargetEndpoint;
import com.ds.infrastructure.hyperspace.common.config.TargetEndpointConfig;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.EndpointInitException;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.channel.EventLoop;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * @author  Weishaoying
 */
@Slf4j
public class NettyTargetEndpoint implements TargetEndpoint<TargetEndpointConfig> {

    private final NettyContainer nettyContainer;

    private volatile TargetEndpointConfig targetEndpointConfig;

    private volatile Map<String, List<EndpointNode<TargetNodeConfig>>> zoneEndpointNodeMap;

    public NettyTargetEndpoint(TargetEndpointConfig targetEndpointConfig,
                               NettyContainer nettyContainer) throws Exception {
        this.targetEndpointConfig = targetEndpointConfig;
        this.nettyContainer = nettyContainer;
        init();
    }

    @Override
    public void init() throws Exception {
        List<TargetNodeConfig> nodes = targetEndpointConfig.getNodes();
        if (CollectionUtils.isEmpty(nodes)) {
            log.warn("Target node is null, endpoint name {}.", targetEndpointConfig.getNamespace());
            this.zoneEndpointNodeMap = null;
            return;
        }

        Map<String, List<EndpointNode<TargetNodeConfig>>> tempZoneEndpointNodeMap = new ConcurrentHashMap<>();
        for (TargetNodeConfig targetNodeConfig : nodes) {
            try {
                if (targetNodeConfig.getProtocolConfig() == null) {
                    targetNodeConfig.setProtocolConfig(nettyContainer.getProtocolConfig());
                }
                TargetEndpointNode tempTargetNode = nettyContainer.getTargetEndpointNode(
                        targetEndpointConfig.getNamespace(), targetNodeConfig);
                if (tempTargetNode != null) {
                    tempTargetNode.updateNodeConfig(targetNodeConfig);
                    tempZoneEndpointNodeMap.computeIfAbsent(targetNodeConfig.getZone().toLowerCase(),
                            key -> new ArrayList<>()).add(tempTargetNode);
                    continue;
                }

                tempTargetNode = new TargetEndpointNode(
                        nettyContainer, targetNodeConfig, this);
                nettyContainer.putTargetEndpointNode(tempTargetNode);
                tempZoneEndpointNodeMap.computeIfAbsent(targetNodeConfig.getZone().toLowerCase(),
                        key -> new ArrayList<>()).add(tempTargetNode);
                initNodeStreamAndManager(targetNodeConfig, tempTargetNode);
            } catch (Exception e) {
                log.error("Init target endpoint node {}:{} failed!",
                        targetNodeConfig.getHost(), targetNodeConfig.getPort(), e);
                throw new EndpointInitException("Init target endpoint node " +
                        targetNodeConfig.getHost() + ":" +
                        targetNodeConfig.getPort().toString() + " failed!");

            }
        }
        this.zoneEndpointNodeMap = tempZoneEndpointNodeMap;
    }

    @Override
    public void reload(TargetEndpointConfig newEndpointConfig) throws Exception {
        TargetEndpointConfig oldEndpointConfig = targetEndpointConfig;
        this.targetEndpointConfig = newEndpointConfig;
        init();
        updateNodeStreamAndManager(oldEndpointConfig, newEndpointConfig);
    }

    @Override
    public List<EndpointNode<TargetNodeConfig>> getEndpointNodes(String zone) {
        if (zone == null || zoneEndpointNodeMap == null) {
            return null;
        }
        return zoneEndpointNodeMap.get(zone);
    }

    @Override
    public TargetEndpointConfig getEndpointConfig() {
        return targetEndpointConfig;
    }

    @Override
    public String getNamespace() {
        return targetEndpointConfig.getNamespace();
    }

    private void initNodeStreamAndManager(TargetNodeConfig nodeConfig, TargetEndpointNode targetEndpointNode) {
        Map<EventLoop, HyperspaceConnector> hyperspaceConnectorMap = nettyContainer.getHyperspaceConnectorMap();
        for (HyperspaceConnector hyperspaceConnector : hyperspaceConnectorMap.values()) {
            hyperspaceConnector.eventLoop().submit(() -> {
                try {
                    hyperspaceConnector.initTargetNodeStreamManager(targetEndpointNode);
                } catch (Exception e) {
                    log.error("Init node stream manager failed!", e);
                    throw new EndpointInitException("Init node stream manager failed!", e);
                }
            });
        }

        if (nodeConfig.getProtocol().isLongLive()) {
            return;
        }
        int connectionInitialPoolSize = nodeConfig.getProtocolConfig().getConnectionInitialPoolSize();
        int currentInitPoolSize = connectionInitialPoolSize / hyperspaceConnectorMap.size();
        if (currentInitPoolSize < 1 && connectionInitialPoolSize > 0) {
            currentInitPoolSize = 1;
        }
        AtomicInteger initedConnectionPoolSize = new AtomicInteger();
        AtomicBoolean isSuccess = new AtomicBoolean(true);
        final TargetEndpointNode lastEndpointNode = targetEndpointNode;
        while (initedConnectionPoolSize.get() < connectionInitialPoolSize) {
            if (!isSuccess.get()) {
                throw new EndpointInitException("Stream init failed!");
            }
            for (HyperspaceConnector hyperspaceConnector : hyperspaceConnectorMap.values()) {
                if (initedConnectionPoolSize.get() >= connectionInitialPoolSize) {
                    break;
                }
                final int tempInitPoolSize = currentInitPoolSize;
                final CountDownLatch countDownLatch = new CountDownLatch(1);
                hyperspaceConnector.eventLoop().submit(() -> {
                    try {
                        hyperspaceConnector.initTargetNodeStream(lastEndpointNode, tempInitPoolSize);
                        initedConnectionPoolSize.addAndGet(tempInitPoolSize);
                    } catch (Exception e) {
                        log.error("Init HyperspaceConnector failed!", e);
                        isSuccess.set(false);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
                try {
                    countDownLatch.await(nettyContainer.getResourceConfig()
                            .getTargetInitMaxTimeout(), MILLISECONDS);
                } catch (InterruptedException e) {
                    log.error("Await init event loop pool failure!", e);
                    throw new EndpointInitException("Await target endpoint pool init failure!", e);
                }
                if (!isSuccess.get()) {
                    break;
                }
            }
            if (initedConnectionPoolSize.get() < connectionInitialPoolSize) {
                currentInitPoolSize = 1;
            }
        }
    }

    private void updateNodeStreamAndManager(TargetEndpointConfig oldEndpointConfig,
                                            TargetEndpointConfig newEndpointConfig) {
        List<TargetNodeConfig> oldNodeList = oldEndpointConfig.getNodes();
        List<TargetNodeConfig> deleteNodeList;
        if (!CollectionUtils.isEmpty(newEndpointConfig.getNodes())) {
            deleteNodeList = new ArrayList<>();
            Map<String, String> newNodeMap = new HashMap<>();
            for (TargetNodeConfig newNodeConfig : newEndpointConfig.getNodes()) {
                String newKey = TargetEndpointNode.getNodeKey(newEndpointConfig.getNamespace(), newNodeConfig);
                newNodeMap.put(newKey, "");
            }
            for (TargetNodeConfig oldNodeConfig : oldNodeList) {
                String oldKey = TargetEndpointNode.getNodeKey(oldEndpointConfig.getNamespace(), oldNodeConfig);
                if (newNodeMap.get(oldKey) == null) {
                    deleteNodeList.add(oldNodeConfig);
                }
            }
        } else {
            deleteNodeList = oldNodeList;
        }

        removeNodeStreamAndManager(deleteNodeList, oldEndpointConfig);
    }

    public void removeNodeStreamAndManager(List<TargetNodeConfig> deleteNodeList,
                                           TargetEndpointConfig oldEndpointConfig) {
        if (CollectionUtils.isEmpty(deleteNodeList)) {
            return;
        }

        for (TargetNodeConfig deleteNodeConfig : deleteNodeList) {
            TargetEndpointNode existsTargetEndpointNode = nettyContainer.getTargetEndpointNode(
                    oldEndpointConfig.getNamespace(), deleteNodeConfig);
            if (existsTargetEndpointNode == null) {
                continue;
            }
            existsTargetEndpointNode.shutdown();
            nettyContainer.removeTargetEndpointNode(existsTargetEndpointNode);
            for (HyperspaceConnector hyperspaceConnector : nettyContainer.getHyperspaceConnectorMap().values()) {
                hyperspaceConnector.eventLoop().submit(() -> {
                    try {
                        hyperspaceConnector.removeTargetNodeStreamManager(existsTargetEndpointNode);
                    } catch (Exception e) {
                        log.error("Remove target node stream manager failed!", e);
                        throw new EndpointInitException("Remove target node stream manager failed!", e);
                    }
                });
            }
        }
    }
}
